package scheduler

import (
	"context"
	"fmt"
	"github.com/gin-gonic/gin"
	"go.uber.org/zap"
	"gorm.io/gorm"
	"jk-time/webook/pkg/ginx"
	"jk-time/webook/pkg/gormx/connpool"
	"jk-time/webook/pkg/logger"
	"jk-time/webook/pkg/migrator"
	"jk-time/webook/pkg/migrator/events"
	"jk-time/webook/pkg/migrator/validator"
	"sync"
	"time"
)

type Scheduler[T migrator.Entity] struct {
	lock    sync.Mutex
	src     *gorm.DB
	dst     *gorm.DB
	pool    *connpool.DoubleWritePool
	pattern string

	producer events.Producer

	cancelFull func()
	cancelIncr func()
	//允许多个全量校验同时运行
	fulls map[string]func()
	l     logger.Logger
}

func NewScheduler[T migrator.Entity](l logger.Logger, src *gorm.DB, dst *gorm.DB,
	pool *connpool.DoubleWritePool, producer events.Producer, pattern string) *Scheduler[T] {
	return &Scheduler[T]{
		l:        l,
		lock:     sync.Mutex{},
		src:      src,
		dst:      dst,
		pool:     pool,
		pattern:  pattern,
		producer: producer,
		cancelFull: func() {

		},
		cancelIncr: func() {

		},
		fulls: make(map[string]func()),
	}
}

func (s *Scheduler[T]) RegisterRoutes(server *gin.RouterGroup) {
	// 将这个暴露为 HTTP 接口
	// 你应该配上对应的 UI
	server.POST("/src_only", ginx.Wrap(s.SrcOnly))
	server.POST("/src_first", ginx.Wrap(s.SrcFirst))
	server.POST("/dst_only", ginx.Wrap(s.DstOnly))
	server.POST("/dst_first", ginx.Wrap(s.DstFirst))
	server.POST("/full/start", ginx.Wrap(s.StartFullValidation))
	server.POST("/full/stop", ginx.Wrap(s.StopFullValidation))
	server.POST("/intr/start", ginx.WrapBody[StartIncrRequest](s.StartIncrmentValidation))
	server.POST("/incr/stop", ginx.Wrap(s.StopIncrmentValidation))
}

// SrcOnly 只读写源表
func (s *Scheduler[T]) SrcOnly(c *gin.Context) (ginx.Result, error) {
	s.lock.Lock()
	defer s.lock.Unlock()
	s.pattern = connpool.PatternSrcOnly
	s.pool.ChangePattern(connpool.PatternSrcOnly)
	return ginx.Result{
		Msg: "OK",
	}, nil
}
func (s *Scheduler[T]) SrcFirst(c *gin.Context) (ginx.Result, error) {
	s.lock.Lock()
	defer s.lock.Unlock()
	s.pattern = connpool.PatternSrcFirst
	s.pool.ChangePattern(connpool.PatternSrcFirst)
	return ginx.Result{
		Msg: "OK",
	}, nil
}
func (s *Scheduler[T]) DstOnly(c *gin.Context) (ginx.Result, error) {
	s.lock.Lock()
	defer s.lock.Unlock()
	s.pattern = connpool.PatternDstOnly
	s.pool.ChangePattern(connpool.PatternDstOnly)
	return ginx.Result{
		Msg: "OK",
	}, nil
}
func (s *Scheduler[T]) DstFirst(c *gin.Context) (ginx.Result, error) {
	s.lock.Lock()
	defer s.lock.Unlock()
	s.pattern = connpool.PatternDstFirst
	s.pool.ChangePattern(connpool.PatternDstFirst)
	return ginx.Result{
		Msg: "OK",
	}, nil
}

// StartFullValidation 全量校验
func (s *Scheduler[T]) StartFullValidation(c *gin.Context) (ginx.Result, error) {
	s.lock.Lock()
	defer s.lock.Unlock()
	//取消上一次
	cancel := s.cancelIncr
	v, err := s.newValidator()
	if err != nil {
		zap.L().Error("新的校验失败:", zap.Error(err))
		return ginx.Result{
			Code: 5,
			Msg:  "系统异常",
		}, nil
	}

	var ctx context.Context
	ctx, s.cancelFull = context.WithCancel(c)

	go func() {
		cancel()
		//开始校验
		v.Validate(ctx)
	}()
	return ginx.Result{
		Msg: "启动全量校验成功",
	}, nil

}

// StopFullValidation 全量校验
func (s *Scheduler[T]) StopFullValidation(c *gin.Context) (ginx.Result, error) {
	s.lock.Lock()
	defer s.lock.Unlock()
	s.cancelFull()
	return ginx.Result{
		Msg: "OK",
	}, nil
}

// StartIncrmentValidation 增量校验
func (s *Scheduler[T]) StartIncrmentValidation(c *gin.Context, req StartIncrRequest) (ginx.Result, error) {
	s.lock.Lock()
	defer s.lock.Unlock()
	//取消上一次
	cancel := s.cancelIncr
	v, err := s.newValidator()
	if err != nil {
		zap.L().Error("新的校验失败:", zap.Error(err))
		return ginx.Result{
			Code: 5,
			Msg:  "系统异常",
		}, nil
	}

	v.SleepInterval(time.Duration(req.Interval) * time.Millisecond).Utime(req.Utime)
	var ctx context.Context
	ctx, s.cancelIncr = context.WithCancel(c)

	go func() {
		cancel()
		//开始校验
		v.Validate(ctx)
	}()
	return ginx.Result{
		Msg: "启动增量校验成功",
	}, nil
}

// StopIncrmentValidation 增量校验
func (s *Scheduler[T]) StopIncrmentValidation(c *gin.Context) (ginx.Result, error) {
	s.lock.Lock()
	defer s.lock.Unlock()
	s.cancelIncr()
	return ginx.Result{
		Msg: "OK",
	}, nil
}
func (s *Scheduler[T]) newValidator() (*validator.Validator[T], error) {
	switch s.pattern {
	case connpool.PatternSrcFirst, connpool.PatternSrcOnly:
		return validator.NewValidator[T](validator.NewBaseValidator(s.src, s.dst, "SRC", s.l, s.producer)), nil
	case connpool.PatternDstFirst, connpool.PatternDstOnly:
		return validator.NewValidator[T](validator.NewBaseValidator(s.src, s.dst, "DST", s.l, s.producer)), nil
	default:
		return nil, fmt.Errorf("未知的 pattern %s", s.pattern)
	}
}

type StartIncrRequest struct {
	Utime int64 `json:"utime"`
	// 毫秒数
	// json 不能正确处理 time.Duration 类型
	Interval int64 `json:"interval"`
}
